/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.processor;

import lk.mrt.cse.wihidum.communicator.*;
import lk.mrt.cse.wihidum.storage.*;
import lk.mrt.cse.wihidum.util.*;
import org.apache.axiom.om.*;
import org.apache.axiom.soap.*;
import org.apache.log4j.Logger;

import javax.xml.namespace.QName;
import java.util.*;

/**
 * Process subscription messages
 */
public class SubscriptionProcessor
        implements Processor
{
    private static final Logger logger = Logger.getLogger(SubscriptionProcessor.class);
   // logger.info("Reading my details");
    HashMap<String, String> myBrokerMap = SelfReader.read();
    Router router;
     /**
     * @see lk.mrt.cse.wihidum.processor.Processor#processMessage(org.apache.axiom.soap.SOAPEnvelope, String)
     */
    public SOAPEnvelope processMessage(SOAPEnvelope subscriptionEnvelope, String messageID)
    {
        try
        {
            Subscriber subscriber;
            SubscriberIdentifier subIdentifier;

            // Get the body element of request SOAP envelop
            SOAPBody subscriptionEnvelopBody = subscriptionEnvelope.getBody();

            //get the "Subscribe" element in the soap body
            OMElement subscribeElement = subscriptionEnvelopBody.getFirstChildWithName(new QName(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.ElementNames.Subscribe));

            // Get notify to address
            OMElement deliveryElement = subscribeElement.getFirstChildWithName(new QName(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.ElementNames.Delivery));
            OMElement notifyToElement = deliveryElement.getFirstChildWithName(new QName(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.ElementNames.NotifyTo));
            OMElement notifyToAddress = notifyToElement.getFirstChildWithName(new QName(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ElementNames.Address));
            String sinkAddress = notifyToAddress.getText();

            //get subscription id from reference properties
            OMElement referencePropertiesElement = notifyToElement.getFirstChildWithName(new QName(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ElementNames.ReferenceProperties));
            OMElement subscriptionIDElement = referencePropertiesElement.getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.ElementNames.SubscriptionID));
            String subscrptiptionID = subscriptionIDElement.getText();

            //get filter topic element
            OMElement filterElement = subscribeElement.getFirstChildWithName(new QName(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.ElementNames.Filter));
            String filterString = filterElement.getText();

            String uuid=UUID.randomUUID().toString();
            //Map<SubscriberIdentifier,Subscriber> subscriptionMap=new HashMap<SubscriberIdentifier,Subscriber>();

            subIdentifier=new SubscriberIdentifier(subscrptiptionID,filterString);
            subscriber=new Subscriber();
            subscriber.setSubscriberIdentifier(subIdentifier);
            subscriber.setUuid(uuid);
            subscriber.setNotifyToAddress(sinkAddress);
            //put subscription in hash map
            boolean savedSuccessfully=SubscriptionMap.saveSubscriber(subIdentifier,subscriber);
           // TODO: according to the boolean value, if it is false, a faulty response should be sent to the faulty channel

            if(savedSuccessfully)
            {
                if((myBrokerMap.get("topology")).equals(WihidumConstants.TopologyNames.singleBroker))
                {
                    logger.info("This is a single broker");

                }
                //route the subscription message
                if(!((myBrokerMap.get("topology")).equals(WihidumConstants.TopologyNames.singleBroker)))
                {
                    MessageRoutingThread routingTread = new MessageRoutingThread();
                    routingTread.setSubscriptionMessage(subscriptionEnvelope);
                    routingTread.setFilter(filterString);
                    routingTread.setLogUUID(messageID);
                    routingTread.setAction(WihidumConstants.Actions.Subscribe);
                    routingTread.start();
                }
              }
             return createSubscriptionResponseEnvelope(subscrptiptionID,uuid,filterString);


        }
        catch (NullPointerException ex)
        {
            logger.error("Error in processing received subscription", ex);
            return createSubscriptionFaultResponseElement();
        }
        catch (OMException ex)
        {
            logger.error("Error in processing received subscription", ex);
            return createSubscriptionFaultResponseElement();
        }
    }

    /**
     * Create response for a valid subscription
     *
     * @return Response element
     */
    private SOAPEnvelope createSubscriptionResponseEnvelope(String subscriptionID,String uuid,String filter)
    {

        SOAPFactory factory = OMAbstractFactory.getSOAP11Factory();
        // Create the response SOAP envelop
        SOAPEnvelope responseEnvelop = factory.getDefaultEnvelope();
        OMNamespace wseventingNS = factory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.EVENTING_PREFIX);
        responseEnvelop.declareNamespace(wseventingNS);
        OMNamespace wsaNS = factory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
        responseEnvelop.declareNamespace(wsaNS);
        OMNamespace wihidumNS = factory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
        responseEnvelop.declareNamespace(wihidumNS);
        SOAPHeader soapHeader = responseEnvelop.getHeader();

        OMElement mySubscriptionIDElement = factory.createOMElement(WSEventingConstants.ElementNames.SubscriptionID, wihidumNS);
        mySubscriptionIDElement.setText(subscriptionID);
        soapHeader.addChild(mySubscriptionIDElement);

        OMElement filterElement=factory.createOMElement(WSEventingConstants.ElementNames.Filter,wihidumNS);
        filterElement.setText(filter);
        soapHeader.addChild(filterElement);

        SOAPBody soapBody = responseEnvelop.getBody();

        OMElement responseElement = factory.createOMElement(WSEventingConstants.ElementNames.SubscribeResponse, wseventingNS);
        //here the current broker acts as the subscription manager.
        OMElement subManagerElement = factory.createOMElement(WSEventingConstants.ElementNames.SubscriptionManager, wseventingNS);
        OMElement subManagerAddressElement = factory.createOMElement(WSEventingConstants.ElementNames.Address, wsaNS);
       // subManagerAddressElement.setText(currentTopology.getMyAddress());
        subManagerAddressElement.setText(myBrokerMap.get("address"));
        subManagerElement.addChild(subManagerAddressElement);

        OMElement referenceParametersElement = factory.createOMElement(WSEventingConstants.ElementNames.ReferenceParameters, wsaNS);
        //identifier element contains global wse:Identifier element which is a unique uuid for this particular broker
        OMElement identifierElement = factory.createOMElement(WSEventingConstants.ElementNames.Identifier, wseventingNS);
        identifierElement.setText(uuid);
        referenceParametersElement.addChild(identifierElement);
        subManagerElement.addChild(referenceParametersElement);
        responseElement.addChild(subManagerElement);

        soapBody.addChild(responseElement);

        return responseEnvelop;
    }

    public SOAPEnvelope processPropogatingMessage(SOAPEnvelope requestEnvelope,String messageID)
    {
        //String filterTopic = null;

        //TO DO : Read the header of the message and get the filter topic and from broker details

        SOAPBody soapBody = requestEnvelope.getBody();

        String fromAddress = soapBody
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "BrokerSubscription"))
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "From"))
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "Address"))
                .getText();
        String fromID=soapBody
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "BrokerSubscription"))
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "From"))
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"ID"))
                .getText();
        String filterTopic = soapBody
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "BrokerSubscription"))
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "Filter"))
                .getText();

        String logID=soapBody
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"BrokerSubscription"))
                .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"LogID"))
                .getText();
        String brokerSubscriptionUUID=UUID.randomUUID().toString();
        // save the subscription as it is received from particlular broekr and return correctly created response
        //make a broker subscription object for the new broker subscription
        BrokerSubscription brokerSubscription = new BrokerSubscription();
        brokerSubscription.setFromBrokerAddress(fromAddress);

        brokerSubscription.setFromBrokerID(fromID);
        brokerSubscription.setTopic(filterTopic);
        //put broker subscription in hashmap
        BrokerSubscriptionMap.saveSubscriber(brokerSubscriptionUUID,brokerSubscription);
        //save broker subscription in the database
       // BrokerSubscriptionManager brokerSubsriptionDAO = new BrokerSubscriptionDAO();
        //brokerSubsriptionDAO.saveBrokerSubscription(brokerSubscription, "Received");

        //logger.info("Broker subscription sent from broker " + fromAddress + " to broker " +/* currentTopology.getMyId()*/ " saved successfully");

        try
        {
            MessageRoutingThread routingTread = new MessageRoutingThread();
            routingTread.setSubscriptionMessage(requestEnvelope);
            routingTread.setFilter(filterTopic);
            routingTread.setLogUUID(logID);
            routingTread.setAction(WihidumConstants.Operations.BROKER_SUBSCRIBE);
            routingTread.start();
            //and handover it to router to send it upward.
            return createBrokerSubscriptionResponse();
        }
        catch (NullPointerException ex)
        {
            logger.error("Error in routing received broker subscription", ex);
            return createSubscriptionFaultResponseElement();
        }
        catch (OMException ex)
        {
            logger.error("Error in routing received broker subscription", ex);
            return createSubscriptionFaultResponseElement();
        }


    }
     /**
     * create response for a broker subscription
     */
    private SOAPEnvelope createBrokerSubscriptionResponse()
    {

        SOAPFactory factory = OMAbstractFactory.getSOAP11Factory();
        SOAPEnvelope brokerSubscriptionResponseEnvelope = factory.getDefaultEnvelope();

        OMNamespace wsaNS = factory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
        brokerSubscriptionResponseEnvelope.declareNamespace(wsaNS);
        OMNamespace wihidumNS = factory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
        brokerSubscriptionResponseEnvelope.declareNamespace(wihidumNS);
        OMElement responseElement = factory.createOMElement(WihidumConstants.MessageElements.BROKER_SUBSCRIBE_RESPONSE, wihidumNS);
        //here the current broker acts as the subscription manager.
        OMElement subManagerElement = factory.createOMElement(WSEventingConstants.ElementNames.SubscriptionManager, wihidumNS);
        OMElement subManagerAddressElement = factory.createOMElement(WSEventingConstants.ElementNames.Address, wsaNS);
        //subManagerAddressElement.setText(currentTopology.getMyAddress());
        subManagerAddressElement.setText(myBrokerMap.get("id"));
        subManagerElement.addChild(subManagerAddressElement);

        OMElement referenceParametersElement = factory.createOMElement(WSEventingConstants.ElementNames.ReferenceParameters, wsaNS);
        //identifier element contains global wse:Identifier element which is a unique uuid for this particular broker
        OMElement identifierElement = factory.createOMElement(WSEventingConstants.ElementNames.Identifier, wihidumNS);
        identifierElement.setText(myBrokerMap.get("address"));
        referenceParametersElement.addChild(identifierElement);
        subManagerElement.addChild(referenceParametersElement);
        responseElement.addChild(subManagerElement);
        brokerSubscriptionResponseEnvelope.getBody().addChild(responseElement);

        return brokerSubscriptionResponseEnvelope;

    }

    /**
     * Create response for an invalid subscription
     *
     * @return Response element
     */
    private SOAPEnvelope createSubscriptionFaultResponseElement()
    {
        return null;
    }

    public class MessageRoutingThread extends Thread
    {
        private SOAPEnvelope subscriptionEnvelope;
        private String filterString;
        private String logUUID;

        public String getLogUUID()
        {
            return logUUID;
        }

        public void setLogUUID(String logUUID)
        {
            this.logUUID = logUUID;
        }

        public void setAction(String action)
        {
            this.action = action;
        }

        private String action;

        public SOAPEnvelope getSubscriptionMessage()
        {
            return subscriptionEnvelope;
        }

        public void setSubscriptionMessage(SOAPEnvelope subscriptionMessage)
        {
            this.subscriptionEnvelope = subscriptionMessage;
        }

        public String getFilter()
        {
            return filterString;
        }

        public void setFilter(String filter)
        {
            this.filterString = filter;
        }

        public void run()
        {
            //HashMap<String, String> brokerMap = (HashMap<String, String>)SelfReader.read();
            if (myBrokerMap.get("topology").equals(WihidumConstants.TopologyNames.clusterTopology))
           // if (WihidumConstants.currentTopology.equals(WihidumConstants.TopologyNames.clusterTopology))
            {
                router = new ClusterRouter();
                router.propagateSubscription(subscriptionEnvelope, filterString, action,logUUID);
            }
            //if (WihidumConstants.currentTopology.equals(WihidumConstants.TopologyNames.treeTopology))
            if (myBrokerMap.get("topology").equals(WihidumConstants.TopologyNames.treeTopology))
            {
                router = new TreeRouter();
                router.propagateSubscription(subscriptionEnvelope, filterString, action,logUUID);
            }
        }
    }

    public SOAPEnvelope processUnsubscribeMessage(SOAPEnvelope unsubscribeRequest)
    {

        try
        {
            // Get the body element of request SOAP envelop
            SOAPHeader unsubscriptionHeader = unsubscribeRequest.getHeader();

            //get the "Subscribe" element in the soap body
            OMElement identifierElement = unsubscriptionHeader.getFirstChildWithName(new QName(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.ElementNames.Identifier));
            String subscriberUUID=identifierElement.getText();

            SubscriptionMap.deleteSubscriber(subscriberUUID);

            SubscriberDAO subsDAO = new SubscriberDAO();
            subsDAO.deleteSubscriber(subscriberUUID);

            return createUnsubscribeResponse();
        }
        catch (NullPointerException ex)
        {
            logger.error("Error in processing received subscription", ex);
            return createSubscriptionFaultResponseElement();
        }
        catch (OMException ex)
        {
            logger.error("Error in processing received subscription", ex);
            return createSubscriptionFaultResponseElement();
        }
       // return unsubscribeResponse;
    }
    private SOAPEnvelope createUnsubscribeResponse()
    {
         SOAPFactory factory=OMAbstractFactory.getSOAP12Factory();
         SOAPEnvelope unsubscriptionResponseEnvelope=factory.getDefaultEnvelope();

        OMNamespace wseventingNS=factory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE,WSEventingConstants.EVENTING_PREFIX);
        unsubscriptionResponseEnvelope.declareNamespace(wseventingNS);
        OMNamespace wsaNS=factory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE,WSEventingConstants.ADDRESSING_PREFIX);
        unsubscriptionResponseEnvelope.declareNamespace(wsaNS);
        OMNamespace wihidumNS=factory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE,WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
        unsubscriptionResponseEnvelope.declareNamespace(wihidumNS);
        SOAPHeader soapHeader=unsubscriptionResponseEnvelope.getHeader();


        SOAPBody soapBody=unsubscriptionResponseEnvelope.getBody();

        OMElement unsubscriptionElement =factory.createOMElement(WSEventingConstants.ElementNames.Unsubscribe, wseventingNS);
        soapBody.addChild(unsubscriptionElement);
        return unsubscriptionResponseEnvelope;
    }
}